package com.lovecaa;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * 拉模式-随机获取queue消息
 * 创建时间： 2023年10月21日 11:41:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class LitePullConsumerAssign {
    public static void main(String[] args) throws MQClientException {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer();
        consumer.setConsumerGroup("LitePullConsumer");
        consumer.setNamesrvAddr("192.168.192.110:9876");
        // 这里必须要先启动，不然获取不到queue
        consumer.start();
        // 获取指定Topic的所有queue
        Collection<MessageQueue> queues = consumer.fetchMessageQueues("TopicTest");
        ArrayList<MessageQueue> messageQueues = new ArrayList<>(queues);
        // 手动将消息队列列表分配给此使用者
        consumer.assign(messageQueues);
        // 指定消费哪个queue种的消息，这是第0个，一次拉取10条进行消费
        consumer.seek(messageQueues.get(0), 10);
        while (true) {
            List<MessageExt> messageExtList = consumer.poll();
            System.out.println("消息拉取成功");
            for (MessageExt messageExt : messageExtList) {
                String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                System.out.println("消息消费成功：" + messageExt);
                System.out.println("消息消费成功：" + body);
            }
        }
    }
}
